package com.irdstudio.bhcredit.core.assembly.plugin.dataexport;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.beanutils.BeanUtils;

import com.irdstudio.bhcredit.agent.utils.HexadecimalUtil;
import com.irdstudio.bhcredit.core.assembly.plugin.AbstractPlugin;
import com.irdstudio.bhcredit.core.assembly.plugin.DaoUtil;
import com.irdstudio.bhcredit.core.assembly.plugin.FileHandleUtil;
import com.irdstudio.bhcredit.core.assembly.plugin.util.PluginUtil;
import com.irdstudio.bhcredit.core.util.pub.PathUtil;

/**
   * 导出数据到文本文件中
 * 
 * @author guangming.li
 * @version 1.0
 * @date 2014-09-16
 * @update by fangmf 修改成多线程处理
 */
public class ExportTxtFilePlugin extends AbstractPlugin {

	/* 表数据装配配置列表 */
	private List<PluginExportConf> exportList = null;

	/* 每200万起一个线程执行 */
	private int records = 1000000;

	/**
	 * 从数据库中读取导出配置列表
	 */
	@Override
	protected boolean doReadConfigureFromDB(Connection conn, String szConfIdentify) throws SQLException {
		PluginExportConfDao pecDao = new PluginExportConfDao(conn);
		exportList = pecDao.queryWithPluginConfId(szConfIdentify);
		if (exportList.size() < 1) {
			context.setSzLastErrorMsg("未读取到配置标识为：" + szConfIdentify + "的数据导出配置!");
			return false;
		} else {
			return true;
		}
	}

	/**
	 * 导出数据到文本文件
	 */
	public boolean execute() {

		boolean flag = true;

		FileWriter fos = null;

		// 循环配置装载配置列表，逐个执行装载
		for (int i = 0; i < exportList.size(); i++) {

			PluginExportConf exportConf = exportList.get(i);

			// 解释数据导出配置中的变量(支持系统变量系统中的变量)
			exportConf.setExportTarget(context.toParseSysVariable(exportConf.getExportTarget()));
			exportConf.setExportToFile(context.toParseSysVariable(exportConf.getExportToFile()));
			exportConf.setOkFileFullName(context.toParseSysVariable(exportConf.getOkFileFullName()));
			exportConf.setOkFileContent(context.toParseSysVariable(exportConf.getOkFileContent()));
			exportConf.setFieldSeparator(HexadecimalUtil.dealHexadecimal(exportConf.getFieldSeparator()));
			this.logger.info("........................................................................");
			this.logger.info(
					"准备导出数据：从\"" + exportConf.getExportTarget() + "\"到\"" + exportConf.getExportToFile() + "\"...");

			// 执行装载
			Connection conn = this.getPluginConnection();
			try {
				// 查询表数据量
				int total = 0;
				// 导出方式
				String exportMode = exportConf.getExportMode();
				// 表名或者sql语句
				String exportTarget = exportConf.getExportTarget();
				String totalSql = "select COUNT(*) as total from ";
				if (DataExportContants.BAT_EXPORT_MODE_01.equals(exportMode)) {
					totalSql += exportTarget;
				} else {
					totalSql += "(" + exportTarget + ") t";
				}
				// 查询总数
				Map<String, String> totalMap = DaoUtil.getFirstRecord(totalSql, null, conn);
				total = Integer.parseInt(totalMap.get("total") == null ? "0" : totalMap.get("total"));
				// 计算需要开启的线程数
				int threadCount = PluginUtil.getThreadAmount(records, total);
				List<Future<Boolean>> futureList = new ArrayList<Future<Boolean>>();
				ExecutorService threadPoll = null;
				List<PluginExportConf> pecList = null;
				if (threadCount > 0) {
					// 使用ExecutorService管理线程池
					logger.info("......启动线程["+threadCount+"条]完成插件["+exportConf.getPluginConfId()+"]导数任务......");
					threadPoll = Executors.newFixedThreadPool(threadCount);
					pecList = threadPluginExportConf(exportConf, threadCount);
					for (PluginExportConf pluginExportConf : pecList) {
						// 执行线程，并把结果放入futureList中,注意获取新的数据库连接
						futureList.add(threadPoll
								.submit(new DataExportCallable(pluginExportConf, this.getPluginConnection(), logger)));
					}
				}
				//检查是否启动了线程，启动了要等待所有线程完成
				if (futureList.size() > 0) {
					// 阻塞等待所有线程执行完成
					for (Future<Boolean> future : futureList) {
						if (!future.get()) {
							throw new Exception("执行["+exportConf.getPluginConfId()+"]线程处理异常！");
						}
					}
					// 关闭线程池
					if (threadPoll != null) {
						threadPoll.shutdown();
					}
				}
				//表记录数为空的情况，需要生成空文件
				if(total == 0) {
					// 创建一个空文件
					String localFileName = exportConf.getExportToFile();
					localFileName = localFileName.replace('\\', '/');
					// 获取文件目录
					String fileDir = localFileName.substring(0, localFileName.lastIndexOf("/"));
					if (!new File(fileDir).isDirectory()) {
						// 如果本地文件夹不存在，尝试建立文件夹
						PathUtil.checkAndCreate(fileDir);
						if (!new File(fileDir).isDirectory()) {
							// 创建文件夹失败，抛出异常
							logger.info("创建本地文件夹[" + fileDir + "]失败");
							throw new Exception("创建本地文件夹[" + fileDir + "]失败");
						}
					}
					fos = new FileWriter(exportConf.getExportToFile());
					fos.close();
				}
				
				// 执行文件合并
				logger.info("插件["+exportConf.getPluginConfId()+"],多线程[" + threadCount + "]执行完成，进行文件合并操作..........");
				long mergeStart = System.currentTimeMillis();

				boolean fileFlag = this.getSplitFilePaths(pecList);
				logger.info("插件["+exportConf.getPluginConfId()+"],文件合并结束[" + fileFlag + "]，用时[" + (System.currentTimeMillis() - mergeStart) / 1000
						+ "]秒..........");
				// 生成OK文件
				String okFileContent = "";
				if (DataExportContants.IS_OK_FILE_YES.equals(exportConf.getIsOkFile())) {
					fos = new FileWriter(exportConf.getOkFileFullName());
					okFileContent = exportConf.getOkFileContent() == null ? ""
							: exportConf.getOkFileContent().replace(DataExportContants.RECORD_NUMBER,
									String.valueOf(total));
					fos.write(okFileContent);
					fos.close();
					logger.info("插件["+exportConf.getPluginConfId()+"],生成OK文件[" + exportConf.getOkFileFullName() + "]结束。");
				}
				logger.info("插件["+exportConf.getPluginConfId()+"],导出记录总数:" + total);
			} catch (Exception e) {
				logger.error("导出出错", e);
				flag = false;
			} finally {
				if (fos != null) {
					try {
						fos.close();
					} catch (IOException e) {
						logger.error("关闭资源出错", e);
					}
				}
			}
			this.closePluginConnection(conn);
		}
		return flag;
	}

	/**
	 * TODO 修改PluginExportConf对象，满足每次线程跑的要求 主要修改文件名称,把查询语句转换成分页查询
	 * 
	 * @param pec         原对象
	 * @param threadCount 开启的线程数
	 * @return 每个线程跑的插件数据配置
	 * @author Mr.Fang
	 * @throws CloneNotSupportedException
	 * @throws InvocationTargetException
	 * @throws IllegalAccessException
	 * @email luck_2088@163.com
	 * @date 2019年1月22日 下午7:56:11
	 */
	public List<PluginExportConf> threadPluginExportConf(PluginExportConf pec, int threadCount)
			throws CloneNotSupportedException, IllegalAccessException, InvocationTargetException {
		List<PluginExportConf> pecList = new ArrayList<PluginExportConf>();
		int satrt = 0, end = records;
		for (int i = 1; i <= threadCount; i++) {
			PluginExportConf pluginExportConf = new PluginExportConf();
			BeanUtils.copyProperties(pluginExportConf, pec);
			// 修改文件名,第一个文件不修改
			if(i != 1) {
				pluginExportConf.setExportToFile(pec.getExportToFile() + ".split" + i);
			}
			String pageSql = " LIMIT " + satrt + "," + end + ";";
			// 修改查询语句
			if (DataExportContants.BAT_EXPORT_MODE_01.equals(pec.getExportMode())) {
				// 将按表导出，改成按语句导出
				pluginExportConf.setExportMode(DataExportContants.BAT_EXPORT_MODE_02);
				pluginExportConf.setExportTarget("SELECT * FROM " + pec.getExportTarget() + pageSql);

			} else {
				// 组装成mysql的分页语句
				pluginExportConf.setExportTarget(pec.getExportTarget().replace(";", " ") + pageSql);
			}
			pecList.add(pluginExportConf);
			// 修改下一次查询开始数据
			satrt = end * i;
		}
		return pecList;
	}

	/**
	 * 
	 * TODO 返回拆分生成文件全路径的集合
	 * 
	 * @param pecList
	 * @return 路径集合
	 * @author Mr.Fang
	 * @throws Exception 
	 * @email luck_2088@163.com
	 * @date 2019年1月23日 下午2:32:53
	 */
	public boolean getSplitFilePaths(List<PluginExportConf> pecList) throws Exception {
		List<String> fileInPtahs = new ArrayList<String>();
		String fileOutPath = "";
		if (pecList != null) {
			//当没有分割文件时
			if(pecList.size() == 1) {
				return true;
			}
			//取第一个文件，后续文件合并到该文件中
			fileOutPath = pecList.get(0).getExportToFile();
			for (int i = 1; i < pecList.size(); i++) {
				fileInPtahs.add(pecList.get(i).getExportToFile());
			}
		}
		boolean flag = true ;
		try {
			flag = FileHandleUtil.mergedFiles(fileInPtahs, fileOutPath);
		} catch (Exception e) {
			flag = false;
			throw new Exception(e);
		}
		
		return flag;
	}
}
